package com.lu.core.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Supplier;

/**
 * @program sigo.b2c.api.marketing-release
 * @description: 线程池配置
 * @author: zhanglu
 * @create: 2021-09-10 14:43:00
 */
@Configuration
public class ThreadPoolConfig {

    @Bean
    public ThreadPoolTaskExecutor systemLogHelperExecutor() {
        //根据ThreadPoolTaskExecutor 创建建线程池
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //设置初始的线程数量
        executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
        //设置最大的线程数量
        executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors());
        //任务队列设置最大 任务数量
        executor.setQueueCapacity(10000);
        //设置 超出初始化线程的 存在时间为60秒
        executor.setKeepAliveSeconds(120);
        //设置 线程前缀
        executor.setThreadNamePrefix("systemLogHelperExecutor-");
        //线程池的饱和策略 我这里设置的是 CallerRunsPolicy 也就是由用调用者所在的线程来执行任务 共有四种
        //AbortPolicy：直接抛出异常，这是默认策略；
        //CallerRunsPolicy：用调用者所在的线程来执行任务；
        //DiscardOldestPolicy：丢弃阻塞队列中靠最前的任务，并执行当前任务；
        //DiscardPolicy：直接丢弃任务；
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        //设置在关闭线程池时是否等待任务完成
        executor.setWaitForTasksToCompleteOnShutdown(true);
        //设置等待终止的秒数
        executor.setAwaitTerminationSeconds(60);
        return executor;
    }

    /**
     * 分任务并行处理 模版
     * @param s1 第一个处理函数，返回结果
     * @param s2 第二个处理函数，返回结果
     * @return  得到s1,s2的返回结果，进一步处理，返回处理后的结果
     */
    public <T, U, V> V concurrentHandleTemplate(ThreadPoolTaskExecutor executor, Supplier<T> s1, Supplier<U> s2, BiFunction<? super T,? super U,? extends V> fn){
        V v = CompletableFuture.supplyAsync(() -> s1.get(), executor)
                .thenCombine
                (CompletableFuture.supplyAsync(() -> s2.get(), executor),
                (res1, res2) -> fn.apply(res1, res2))
                .join();
        return v;
    }

    /**
     * 分任务并行处理 模版
     * @param s1 第一个处理函数，返回结果
     * @param s2 第二个处理函数，返回结果
     * @return  得到s1,s2的返回结果，进一步处理，返回处理后的结果
     */
    public <T, U, V> void concurrentHandleTemplate(ThreadPoolTaskExecutor executor, Supplier<T> s1, Supplier<U> s2, BiConsumer<? super T, ? super U> action){
        CompletableFuture.supplyAsync(() -> s1.get(), executor)
                .thenAcceptBoth
                (CompletableFuture.supplyAsync(() -> s2.get(), executor),
                (res1, res2) -> action.accept(res1, res2))
                .join();
    }

}
